package com.usian.article.handler;

import com.alibaba.fastjson.JSON;
import com.usian.article.config.KafkaStreamListener;
import com.usian.model.common.constants.HotArticleConstants;
import com.usian.model.mess.app.ArticleVisitStreamMess;
import com.usian.model.mess.app.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
  * 功能描述：文章分值流式计算
  * @param:
  * @return：
  * @auther：
  */
@Component
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {

    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        //计算
        return stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            /**
             *  1124234:LIKES
             *  1124234:LIKES
             *  1124234:LIKES
             *  1124234:VIEWS
             *  1124235:VIEWS
             *  1124235:LIKES
             *
             *  1124234:LIKES  3
             *  1124234:VIEWS  1
             *  1124235:VIEWS 1
             *  1124235:LIKES  1
             *
             *
             *  1124234 : 点赞3次  阅读1次
             *  1124235 ： 点赞1次  阅读1次
             *
             * @param value
             * @return
             */
            @Override
            public Iterable<String> apply(String value) {
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                return Arrays.asList(new StringBuilder(mess.getArticleId().toString()).append(":").append(mess.getType().name()).toString());
            }
        }).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 把消息的key,重新赋值，目前消息的key就是单词
             * @param key
             * @param value
             * @return
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                return new KeyValue<>(value, value);
            }
        }).groupByKey()
                //时间聚合窗口
                .windowedBy(TimeWindows.of(10000))
                //消息的value就是聚合单词后的统计数值，long类型
                .count(Materialized.as("count-word-num-0008"))
                //转换为Kstream
                .toStream()
                //把处理后的key和value转换String
                .map((key,value)->{
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(), value));
                });
    }

    private String formatObj(String key, Long value) {
        String[] spt = key.split(":");

        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();

        mess.setArticleId(Long.valueOf(spt[0]));
        if (spt[1].equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())) {
            mess.setCollect(value);
        }

        if (spt[1].equals(UpdateArticleMess.UpdateArticleType.LIKES.name())) {
            mess.setLike(value);
        }

        if (spt[1].equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())) {
            mess.setComment(value);
        }

        if (spt[1].equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())) {
            mess.setView(value);
        }
        return JSON.toJSONString(mess);
    }
}
